package com.whut.util.concurrent.threadpool;

import com.whut.util.concurrent.locks.MyReentrantLock;
import com.whut.util.concurrent.threadpool.exception.ThreadPoolException;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author wangzonghe
 * @date 2025/8/11 10:40
 */
@Slf4j
public class MyThreadPool {

    /**
     * 线程池核心线程数
     */
    private final int corePoolSize;

    /**
     * 线程池最大线程数
     */
    private final int maxPoolSize;

    /**
     * 线程池阻塞队列
     */
    private final LinkedBlockingQueue<Runnable> workQueue;

    /**
     * 线程池阻塞队列超时时间
     */
    private final long timeout;

    /**
     * 线程池阻塞队列超时时间单位
     */
    private final TimeUnit unit;

    /**
     * 核心线程列表
     */
    private final List<Thread> coreThreadList;

    /**
     * 辅助线程列表
     */
    private final List<Thread> supportThreadList;

    /**
     * 拒绝策略
     */
    private final RejectHandle rejectHandle;

    /**
     * 线程池锁
     */
    private final MyReentrantLock myReentrantLock;

    private static final long DEFAULT_TIME_OUT = 1;

    private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.SECONDS;

    public MyThreadPool(int corePoolSize, int maxPoolSize, LinkedBlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maxPoolSize, DEFAULT_TIME_OUT, DEFAULT_TIME_UNIT, workQueue, new DefaultRejectHandle());
    }

    public MyThreadPool(int corePoolSize, int maxPoolSize, long timeout, TimeUnit unit, LinkedBlockingQueue<Runnable> workQueue, RejectHandle rejectHandle) {
        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;
        this.workQueue = workQueue;
        this.timeout = timeout;
        this.unit = unit;
        this.rejectHandle = rejectHandle;

        int supportPoolSize = Math.max(maxPoolSize - corePoolSize, 0);
        this.coreThreadList = new ArrayList<>(corePoolSize);
        this.supportThreadList = new ArrayList<>(supportPoolSize);
        this.myReentrantLock = new MyReentrantLock();
    }

    public MyThreadPool(int corePoolSize, int maxPoolSize, long timeout, TimeUnit unit, LinkedBlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maxPoolSize, timeout, unit, workQueue, new DefaultRejectHandle());
    }

    public void execute(Runnable runnable) {

        // 如果核心线程列表小于核心线程数，则创建核心线程
        if (coreThreadList.size() < corePoolSize) {
            addCoreThread();
        }

        // 把任务塞到阻塞队列中，塞进去了就说明放入成功
        if (workQueue.offer(runnable)) {
            return;
        }

        // 阻塞队列满了，则创建辅助线程
        if (supportThreadList.size() < maxPoolSize - corePoolSize && addSupportThread(runnable)) {
            // 增加辅助线程，让辅助线程先执行runnable，然后再去阻塞队列取
            return;
        }

        // 如果阻塞队列还是满的，则执行拒绝策略
        if (!workQueue.offer(runnable)) {
            rejectHandle.reject(runnable, this);
        }

    }

    private boolean addSupportThread(Runnable runnable) {
        // 加锁保障线程安全
        myReentrantLock.unReentrantLockFair();
        try {
            // double check判断辅助线程列表是否小于辅助线程数，小于则创建辅助线程
            if (supportThreadList.size() < maxPoolSize - corePoolSize) {
                Thread supportThread = new SupportThread(runnable);
                supportThread.start();
                supportThreadList.add(supportThread);
                return true;
            }
        } finally {
            myReentrantLock.unlock();
        }
        return false;
    }

    private void addCoreThread() {
        // 加锁保障线程安全
        myReentrantLock.unReentrantLockFair();
        try {
            // double check判断核心线程列表是否小于核心线程数，小于则创建核心线程
            if (coreThreadList.size() < corePoolSize) {
                Thread coreThread = new CoreThread();
                coreThreadList.add(coreThread);
                coreThread.start();
            }
        } finally {
            myReentrantLock.unlock();
        }
    }


    private class CoreThread extends Thread {
        @Override
        public void run() {
            log.info("{} - 核心线程启动", Thread.currentThread().getName());
            while (true) {
                try {
                    Runnable runnable = workQueue.take();
                    runnable.run();
                } catch (InterruptedException e) {
                    log.warn("核心线程被中断", e);
                    Thread.currentThread().interrupt();
                    break;
                }

            }
        }
    }

    private class SupportThread extends Thread {

        private Runnable r;

        public SupportThread(Runnable runnable) {
            this.r = runnable;
        }

        public SupportThread() {

        }

        @Override
        public void run() {
            log.info("{} - 辅助线程创建", Thread.currentThread().getName());
            if (Objects.nonNull(r)) {
                r.run();
            }
            while (true) {
                try {
                    Runnable runnable = workQueue.poll(timeout, unit);
                    if (Objects.isNull(runnable)) {
                        break;
                    }
                    runnable.run();
                } catch (InterruptedException e) {
                    log.warn("核心线程被中断", e);
                    Thread.currentThread().interrupt();
                    throw new ThreadPoolException("辅助线程异常", e);
                }
            }
            log.info("{} - 辅助线程结束", Thread.currentThread().getName());

        }
    }

    private static class DefaultRejectHandle implements RejectHandle {
        @Override
        public void reject(Runnable runnable, MyThreadPool threadPool) {
            throw new ThreadPoolException("线程池已满，无法执行任务");
        }
    }
}
